package org.apache.dolphinscheduler.api.service.impl;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.UUID;
import java.util.stream.Collectors;

import org.apache.commons.lang3.StringUtils;
import org.apache.dolphinscheduler.api.dto.model.ModelFileDto;
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.exceptions.ServiceException;
import org.apache.dolphinscheduler.api.metrics.ApiServerMetrics;
import org.apache.dolphinscheduler.api.service.ModelFileService;
import org.apache.dolphinscheduler.api.service.ResourcesService;
import org.apache.dolphinscheduler.api.utils.RegexUtils;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.FileUtils;
import org.apache.dolphinscheduler.common.utils.PropertyUtils;
import org.apache.dolphinscheduler.dao.entity.Model;
import org.apache.dolphinscheduler.dao.entity.ModelFile;
import org.apache.dolphinscheduler.dao.entity.Tenant;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.ModelFileMapper;
import org.apache.dolphinscheduler.dao.mapper.ModelMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
import org.apache.dolphinscheduler.plugin.storage.api.StorageEntity;
import org.apache.dolphinscheduler.plugin.storage.api.StorageOperate;
import org.apache.dolphinscheduler.spi.enums.ResourceType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.io.Files;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@Service
public class ModelFileServiceImpl extends ServiceImpl<ModelFileMapper, ModelFile> implements ModelFileService {
	
	@Autowired(required = false)
    private StorageOperate storageOperate;
	
	@Autowired
    private TenantMapper tenantMapper;
	
	@Autowired
	private ModelMapper modelMapper;

    @Autowired
    ResourcesService resourcesService;

	public List<ModelFile> queryModelFileByModelId(Integer modelId, Integer pid) {
		QueryWrapper<ModelFile> qw = new QueryWrapper<ModelFile>();
		qw.eq("model_id", modelId);
		if(pid != null) {
			qw.eq("pid", pid);
		}
		List<ModelFile> list = list(qw);
		if(list != null && list.size() > 0) {
			for(ModelFile mf : list) {
				mf.setChildFile(queryModelFileByModelId(modelId, mf.getPid()));
			}
		}
		return list;
	}
	
	public Result<Object> uploadFile(User loginUser, Integer modelId, ModelFileDto dto, MultipartFile file) {
		Result<Object> result = new Result<>();
		Model model = modelMapper.selectById(modelId);
		if(model == null) {
			return Result.error();
		}
		dto.setModelId(modelId);
        result = checkResourceUploadStartupState();
        if (!result.getCode().equals(Status.SUCCESS.getCode())) {
            return result;
        }
        String currDirNFileName = "";
    	if(dto.getPid() != -1) {
    		// 获取父路径
    		ModelFile mf = getById(dto.getPid());
    		if(mf != null) {
    			currDirNFileName = mf.getFullName() + Constants.FOLDER_SEPARATOR + dto.getName();
    		}
    	} else {
    		String tenantCode = getTenantCode(loginUser.getTenantId());
    		currDirNFileName = storageOperate.getResDir(tenantCode) + loginUser.getUserName() + Constants.FOLDER_SEPARATOR + model.getName() + Constants.FOLDER_SEPARATOR + dto.getName();
    	}
        if (checkResourceExists(currDirNFileName)) {
            log.error("resource {} has exist, can't recreate", RegexUtils.escapeNRT(dto.getName()));
            return Result.error(Status.RESOURCE_EXIST);
        }
        ModelFile mfile = new ModelFile();
        if(dto.getIsDirectory() == 0) {
        	// 上传文件
        	result = verifyFile(dto.getName(), file);
            if (!result.getCode().equals(Status.SUCCESS.getCode())) {
                return result;
            }
            if (currDirNFileName.length() > Constants.RESOURCE_FULL_NAME_MAX_LENGTH) {
                log.error(
                        "Resource file's name is longer than max full name length, fullName:{}, " +
                                "fullNameSize:{}, maxFullNameSize:{}",
                        RegexUtils.escapeNRT(dto.getName()), currDirNFileName.length(), Constants.RESOURCE_FULL_NAME_MAX_LENGTH);
                return Result.error(Status.RESOURCE_FULL_NAME_TOO_LONG_ERROR);
            }
            if(upload(loginUser, currDirNFileName, file)) {
            	ApiServerMetrics.recordApiResourceUploadSize(file.getSize());
            	mfile.setSize(file.getSize());
            } else {
            	 return Result.error(Status.STORE_OPERATE_CREATE_ERROR);
            }
        } else {
        	// 创建文件夹
        	result = createDirectory(loginUser, currDirNFileName);
        	if (!result.getCode().equals(Status.SUCCESS.getCode())) {
                return result;
            }
        	mfile.setSize(0L);
        	
        }
        mfile.setModelId(dto.getModelId());
    	mfile.setCreateTime(new Date());
    	mfile.setFileName(dto.getName());
    	mfile.setFileDesc(dto.getRemark());
    	mfile.setFullName(currDirNFileName);
    	mfile.setPid(dto.getPid());
    	mfile.setUid(loginUser.getId());
    	mfile.setIsDirectory(dto.getIsDirectory());
    	save(mfile);
		return Result.success();
	}
	
	public Result<Object> deleteFile(User loginUser, Integer modelId, Integer fileId) {
		ModelFile mf = getById(fileId);
		if(mf == null) {
			return Result.error();
		}
		// 查询删除的文件是否有发布的模型
		boolean flag = countByFullName(mf.getFullName());
		if(flag) {
			return Result.error(Status.EXIST_PUBLISH_NOT_DELETE);
		}
		String tenantCode = getTenantCode(loginUser.getTenantId());
		String defaultPath = storageOperate.getResDir(tenantCode);
        StorageEntity resource = null;
        String fullName = mf.getFullName();
        try {
            resource = storageOperate.getFileStatus(fullName, defaultPath, tenantCode, null);
        } catch (Exception e) {
            log.error(e.getMessage() + " Resource path: {}", fullName, e);
            return Result.error(Status.RESOURCE_NOT_EXIST);
        }
        if (resource == null) {
            log.error("Resource does not exist, resource full name:{}.", fullName);
            return Result.error(Status.RESOURCE_NOT_EXIST);
        }
        List<String> allChildren = storageOperate.listFilesStatusRecursively(fullName, defaultPath,
                tenantCode, resource.getType()).stream().map(storageEntity -> storageEntity.getFullName())
                .collect(Collectors.toList());
        try {
			storageOperate.delete(fullName, allChildren, true);
			// 删除表数据
			removeById(mf);
		} catch (IOException e) {
			log.error("delete file error", e);
			return Result.error(Status.DELETE_RESOURCE_ERROR);
		}
		return Result.success();
	}
	
	public Result<Object> createFile(User loginUser, Integer modelId, ModelFileDto dto) {
		Model model = modelMapper.selectById(modelId);
		if(model == null) {
			return Result.error();
		}
		dto.setModelId(modelId);
		if(StringUtils.isBlank(dto.getContent()) 
				|| StringUtils.isBlank(dto.getName())
				|| StringUtils.isBlank(dto.getFileType())) {
			return Result.error();
		}
		String currDirNFileName = "";
		String tenantCode = getTenantCode(loginUser.getTenantId());
		String fileName = dto.getName() + dto.getFileType();
    	if(dto.getPid() != -1) {
    		// 获取父路径
    		ModelFile mf = getById(dto.getPid());
    		if(mf != null) {
    			currDirNFileName = mf.getFullName() + Constants.FOLDER_SEPARATOR + dto.getName();
    		}
    	} else {
    		currDirNFileName = storageOperate.getResDir(tenantCode) + loginUser.getUserName() + Constants.FOLDER_SEPARATOR + model.getName() + Constants.FOLDER_SEPARATOR + fileName;
    	}
		String localFilename = FileUtils.getUploadFilename(tenantCode, fileName);
		boolean flag = FileUtils.writeContent2File(dto.getContent(), localFilename);
		if(flag) {
			try {
				File file = new File(localFilename);
				storageOperate.upload(tenantCode, localFilename, currDirNFileName, true, true);
				ModelFile mfile = new ModelFile();
            	mfile.setModelId(dto.getModelId());
            	mfile.setCreateTime(new Date());
            	mfile.setFileName(dto.getName());
            	mfile.setFileDesc(dto.getRemark());
            	mfile.setFullName(currDirNFileName);
            	mfile.setSize(file.length());
            	mfile.setPid(dto.getPid());
            	mfile.setUid(loginUser.getId());
            	mfile.setIsDirectory(dto.getIsDirectory());
            	save(mfile);
			} catch (IOException e) {
				log.error("create file error", e);
				return Result.error();
			} finally {
				FileUtils.deleteFile(localFilename);
			}
		} else {
			return Result.error();
		}
		return Result.success();
	}
	
	private boolean countByFullName(String fullName) {
		QueryWrapper<ModelFile> qw = new QueryWrapper<ModelFile>();
		qw.eq("full_name", fullName);
		List<ModelFile> list = new ArrayList<ModelFile>();
		if(list != null && list.size() > 0) {
			for(ModelFile mf : list) {
				Model model = modelMapper.selectById(mf.getModelId());
				if(model != null && model.getStatus() == 2) {
					return true;
				}
			}
		}
		return false;
	}
	
	private Result<Object> checkResourceUploadStartupState() {
        if (!PropertyUtils.isResourceStorageStartup()) {
            log.error("Storage does not start up, resource upload startup state: {}.",
                    PropertyUtils.isResourceStorageStartup());
            return Result.error(Status.STORAGE_NOT_STARTUP);
        }
        return Result.success();
    }
	
	private boolean upload(User loginUser, String fullName, MultipartFile file) {
        // query tenant
        String tenantCode = getTenantCode(loginUser.getTenantId());
        // random file name
        String localFilename = FileUtils.getUploadFilename(tenantCode, UUID.randomUUID().toString());

        // save file to hdfs, and delete original file
        String resourcePath = storageOperate.getDir(ResourceType.FILE, tenantCode);
        try {
            // if tenant dir not exists
            if (!storageOperate.exists(resourcePath)) {
                storageOperate.createTenantDirIfNotExists(tenantCode);
            }
            org.apache.dolphinscheduler.api.utils.FileUtils.copyInputStreamToFile(file, localFilename);
            storageOperate.upload(tenantCode, localFilename, fullName, true, true);
        } catch (Exception e) {
            FileUtils.deleteFile(localFilename);
            log.error(e.getMessage(), e);
            return false;
        }
        return true;
    }
	
	private String getTenantCode(int tenantId) {
        Tenant tenant = tenantMapper.queryById(tenantId);
        if (tenant == null) {
            throw new ServiceException(Status.CURRENT_LOGIN_USER_TENANT_NOT_EXIST);
        }
        return tenant.getTenantCode();
    }
	
	private Result<Object> verifyFile(String name, MultipartFile file) {

        if (FileUtils.directoryTraversal(name)) {
            log.warn("Parameter file alias name verify failed, fileAliasName:{}.", RegexUtils.escapeNRT(name));
            return Result.error(Status.VERIFY_PARAMETER_NAME_FAILED);
        }

        if (file != null && FileUtils.directoryTraversal(Objects.requireNonNull(file.getOriginalFilename()))) {
            log.warn("File original name verify failed, fileOriginalName:{}.",
                    RegexUtils.escapeNRT(file.getOriginalFilename()));
            return Result.error(Status.VERIFY_PARAMETER_NAME_FAILED);
        }

        if (file != null) {
            // file is empty
            if (file.isEmpty()) {
                log.warn("Parameter file is empty, fileOriginalName:{}.",
                        RegexUtils.escapeNRT(file.getOriginalFilename()));
                return Result.error(Status.RESOURCE_FILE_IS_EMPTY);
            }

            if (file.getSize() > Constants.MAX_FILE_SIZE) {
                log.warn(
                        "Resource file size is larger than max file size, fileOriginalName:{}, fileSize:{}, maxFileSize:{}.",
                        RegexUtils.escapeNRT(file.getOriginalFilename()), file.getSize(), Constants.MAX_FILE_SIZE);
                return Result.error(Status.RESOURCE_SIZE_EXCEED_LIMIT);
            }
        }
        return Result.success();
    }
	
	private boolean checkResourceExists(String fullName) {
        Boolean existResource = false;
        try {
            existResource = storageOperate.exists(fullName);
        } catch (IOException e) {
            log.error("error occurred when checking resource: " + fullName, e);
        }
        return Boolean.TRUE.equals(existResource);
    }
	
	private Result<Object> createDirectory(User loginUser, String fullName) {
		String tenantCode = getTenantCode(loginUser.getTenantId());
        String resourceRootPath = storageOperate.getDir(ResourceType.FILE, tenantCode);
        try {
            if (!storageOperate.exists(resourceRootPath)) {
                storageOperate.createTenantDirIfNotExists(tenantCode);
            }

            if (!storageOperate.mkdir(tenantCode, fullName)) {
                log.error("create resource directory {} failed", fullName);
                return Result.error(Status.STORE_OPERATE_CREATE_ERROR);
            }
        } catch (Exception e) {
            log.error("create resource directory {} failed", fullName);
            return Result.error(Status.STORE_OPERATE_CREATE_ERROR);
        }
        return Result.success();
    }

    public Result<Object> downloadFile(User loginUser, Integer modelId) {
    	Model model = modelMapper.selectById(modelId);
		if(model == null) {
			return Result.error();
		}
		if(model.getStatus() == 1) {
			return Result.error(Status.MODEL_NOT_PUBLISH);
		}
        //get model list
        List<ModelFile> fileList = queryModelFileByModelId(modelId, -1);
        if(fileList == null) {
            log.error("resource not exist");
            return Result.error(Status.MODEL_NOT_FILE);
        } else {
        	if(fileList.size() == 0) {
        		return Result.error(Status.MODEL_NOT_FILE);
        	}
        }
        String modelPath = FileUtils.getDownloadFilename(model.getName());
        // download all file
        downloadAllFile(loginUser, fileList, modelPath);
        // zip file
        String zipFileName = modelPath + Constants.FOLDER_SEPARATOR + DateUtils.getCurrentTime("yyyyMMddHHmmss") + ".zip";
        try {
            File dir = new File(modelPath);
            FileUtils.zipDirectory(dir, zipFileName);
        } catch (Exception e) {
            log.error("zip model file {} failed", modelId);
            return Result.error();
        }
        return Result.success(zipFileName);
    }

    private void downloadAllFile(User loginUser, List<ModelFile> fileList, String sourcePath) {
    	String tenantCode = getTenantCode(loginUser.getTenantId());
    	for(ModelFile mf : fileList) {
    		if(mf.getIsDirectory() == 1) { //folder
                String folderPath = sourcePath + Constants.FOLDER_SEPARATOR + mf.getFileName();
                File file = new File(folderPath);
                if (!file.getParentFile().exists()) {
                    file.getParentFile().mkdirs();
                }
        	} else { // file
                String filePath = sourcePath + Constants.FOLDER_SEPARATOR + mf.getFileName();
                try {
                	storageOperate.download(tenantCode, mf.getFullName(), filePath, true);
                } catch (Exception e){
                    log.error("download model file {} failed", mf.getFileName());
                }
            }
            if(mf.getChildFile() != null && mf.getChildFile().size() > 0){
                downloadAllFile(loginUser, mf.getChildFile(), sourcePath + Constants.FOLDER_SEPARATOR + mf.getFileName());
            }
    	}
    }
}
